package com.gi.product.controller;

import com.alibaba.fastjson.JSON;
import com.gi.product.entity.PortraitImage;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * 官方文档地址
 * https://www.elastic.co/guide/en/elasticsearch/client/index.html
 */
@RestController
@RequestMapping("/es")
public class EsRestClientController {
    private static final Logger logger = LoggerFactory.getLogger(EsRestClientController.class);
    @Autowired
    private RestHighLevelClient client;

    private BulkProcessor bulkProcessor;

    /**
     * client初始化
     */
//    @PostConstruct
//    public void init() {
//        try {
//            client = new RestHighLevelClient(
//                    RestClient.builder(
//                            new HttpHost("localhost", 9200, "http")));
//            //new HttpHost("localhost", 9201, "http")));
//        } catch (Exception e) {
//            logger.error("es init failed!", e);
//        }
//    }
    @PostConstruct
    public void init() {
        BulkProcessor.Builder builder = BulkProcessor.builder(
                (request, bulkListener) ->
                        client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
                new BulkProcessor.Listener() {
                    @Override
                    public void beforeBulk(long executionId, BulkRequest request) {
                        int numberOfActions = request.numberOfActions();
                        logger.debug("Executing bulk [{}] with {} requests",
                                executionId, numberOfActions);
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request,
                                          BulkResponse response) {
                        if (response.hasFailures()) {
                            logger.warn("Bulk [{}] executed with failures", executionId);
                            logger.warn("Bulk response [{}] executed with failures", response.toString());
                            logger.warn("Bulk request [{}] executed with failures", request.toString());
                        } else {
                            logger.debug("Bulk [{}] completed in {} milliseconds",
                                    executionId, response.getTook().getMillis());
                        }
                    }

                    @Override
                    public void afterBulk(long executionId, BulkRequest request,
                                          Throwable failure) {
                        logger.error("Failed to execute bulk,，，{}", request.toString());
                    }
                });

        bulkProcessor = builder// 2000条数据请求执行一次bulk
                .setBulkActions(2000)
                // 5mb的数据刷新一次bulk
                .setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB))
                // 并发请求数量, 0不并发, 1并发允许执行
                .setConcurrentRequests(0)
                // 固定5s必须刷新一次
                .setFlushInterval(TimeValue.timeValueSeconds(5L))
                // 重试3次，间隔1s
                .setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3)).build();
    }

    @GetMapping("/sync/index1")
    public IndexResponse test1() {

        Map<String, Object> jsonMap = new HashMap<>();
        jsonMap.put("name", "hello k");
        jsonMap.put("age", 3);
        jsonMap.put("birthday", new Date());
        //Document source provided as a Map which gets automatically converted to JSON format
        //IndexRequest indexRequest = new IndexRequest("cat").id("1").source(jsonMap);

        IndexRequest request = new IndexRequest("cat");
        request.id("1");
        request.source(JSON.toJSONString(jsonMap), XContentType.JSON);
        //同步的方式
        IndexResponse indexResponse = null;
        try {
            indexResponse = client.index(request, RequestOptions.DEFAULT);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        //close
        //client.close();
        return indexResponse;
    }

    @GetMapping("/async/index2")
    public String test2() {
        Map<String, Object> jsonMap = new HashMap<>();
        jsonMap.put("name", "My k2");
        jsonMap.put("age", 2);
        jsonMap.put("birthday", new Date());

        IndexRequest request = new IndexRequest("cat");
        request.id("2");
        request.source(JSON.toJSONString(jsonMap), XContentType.JSON);
        //异步的方式
        client.indexAsync(request, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                logger.info("success {}", indexResponse.getResult());
            }

            @Override
            public void onFailure(Exception e) {
                logger.error("fail {}", e.getMessage());
            }
        });
        return "sync";
    }

    /**
     * 根据id获取单个文档
     */
    @GetMapping("/sync/get/{id}")
    public Map<String, Object> get(@PathVariable String id) throws IOException {
        GetRequest request = new GetRequest("cat", id);
        Map<String, Object> sourceAsMap = null;
        try {
            GetResponse getResponse = client.get(request, RequestOptions.DEFAULT);
            String index = getResponse.getIndex();
            String resId = getResponse.getId();
            if (getResponse.isExists()) {
                long version = getResponse.getVersion();
                String sourceAsString = getResponse.getSourceAsString();
                sourceAsMap = getResponse.getSourceAsMap();
                byte[] sourceAsBytes = getResponse.getSourceAsBytes();
            } else {
                logger.error("没有数据");
            }
        } catch (ElasticsearchException e) {//并没有抛出异常
            if (e.status() == RestStatus.NOT_FOUND) {
                logger.error("未找到数据");
            }
        }

        return sourceAsMap;
    }

    /**
     * 根据id删除单个文档
     */
    @GetMapping("/delete/{id}")
    public ReplicationResponse.ShardInfo delete(@PathVariable String id) throws IOException {
        DeleteRequest request = new DeleteRequest("cat", id);
        DeleteResponse deleteResponse = client.delete(request, RequestOptions.DEFAULT);

        String index = deleteResponse.getIndex();
        String resId = deleteResponse.getId();
        long version = deleteResponse.getVersion();
        ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {

        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure :
                    shardInfo.getFailures()) {
                String reason = failure.reason();
                logger.error("delete id={} failure,reason {}", id, reason);
            }
        }

        if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
            logger.error("id={} NOT_FOUND", id);
        }
        return shardInfo;
    }

    /**
     * 更新单个文档
     *
     * @param index
     * @param id
     * @throws IOException
     */
    @GetMapping("/update/{index}/{id}")
    public String update(@PathVariable String index, @PathVariable String id) throws IOException {
        Map<String, Object> jsonMap = new HashMap<>();
        jsonMap.put("name", "My k2");
        jsonMap.put("age", 4);
        UpdateRequest request = new UpdateRequest(index, id)
                .doc(jsonMap);

        UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        String resindex = updateResponse.getIndex();
        String resid = updateResponse.getId();
        long version = updateResponse.getVersion();
        if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {

        }

        GetResult result = updateResponse.getGetResult();
        String res = "success";
        if (result.isExists()) {
            String sourceAsString = result.sourceAsString();
            Map<String, Object> sourceAsMap = result.sourceAsMap();
            byte[] sourceAsBytes = result.source();
        } else {
            res = "not found";
        }

        return res;
    }

    /**
     * 查询
     *
     * @throws IOException
     */
    @GetMapping("/search/1")
    public SearchHit[] search() throws IOException {
        SearchRequest searchRequest = new SearchRequest("cat", "customer");
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        RestStatus status = searchResponse.status();
        TimeValue took = searchResponse.getTook();
        Boolean terminatedEarly = searchResponse.isTerminatedEarly();
        boolean timedOut = searchResponse.isTimedOut();
        SearchHits hits = searchResponse.getHits();
        Aggregations aggregations = searchResponse.getAggregations();
        TotalHits totalHits = hits.getTotalHits();
// the total number of hits, must be interpreted in the context of totalHits.relation
        long numHits = totalHits.value;
// whether the number of hits is accurate (EQUAL_TO) or a lower bound of the total (GREATER_THAN_OR_EQUAL_TO)
        TotalHits.Relation relation = totalHits.relation;
        float maxScore = hits.getMaxScore();
        SearchHit[] searchHits = hits.getHits();
        for (SearchHit hit : searchHits) {
            String index = hit.getIndex();
            String id = hit.getId();
            float score = hit.getScore();
            String sourceAsString = hit.getSourceAsString();
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
        }

        return hits.getHits();
    }

    /**
     * 查询
     * #搜索单个index
     * GET /bank,cat/_search?q=firstname:Hattie
     * <p>
     * #搜索多个index，用逗号隔开
     * GET /bank,cat/_search?q=name:k
     * <p>
     * #搜索全部index
     * GET /_all/_search?q=name:k
     *
     * @throws IOException
     */
    @GetMapping("/search/2/{name}")
    public SearchHit[] search2(@PathVariable String name) throws IOException {
        // 指定只能在哪些index中查询：可以添加多个且没有限制，中间用逗号隔开
        SearchRequest searchRequest = new SearchRequest("cat", "customer");
        //设置指定查询的路由分片
        //searchRequest.routing("routing");

        //用preference方法去指定优先去某个分片上去查询（默认的是随机先去某个分片）
        //searchRequest.preference("_local");

        //设置过滤条件
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        //查询所有的内容
        //sourceBuilder.query(QueryBuilders.matchAllQuery());

        //查询包含关键词字段的文档：如下，表示查询出来所有包含name字段且user字段包含${name}值的文档
        sourceBuilder.query(QueryBuilders.termQuery("name", name));
        sourceBuilder.from(0);
        sourceBuilder.size(5);
        //设置查询请求的超时时间：如下表示60秒没得到返回结果时就认为请求已超时
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();

        return hits.getHits();
    }

    /**
     * GET /cat/_search
     * {
     * "query":{
     * "match":{
     * "name":"k"
     * }
     * },
     * "highlight": {
     * "fields": {
     * "name": {}
     * }
     * }
     * }
     * <p>
     * 模糊查询，带高亮显示
     *
     * @throws IOException
     */
    @GetMapping("/search/3/{name}/{pageNum}/{pageSize}")
    public List<Map<String, Object>> search3(@PathVariable String name, @PathVariable int pageNum, @PathVariable int pageSize) throws IOException {
        SearchRequest searchRequest = new SearchRequest("cat");
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.termQuery("name", name));

        HighlightBuilder highlightBuilder = new HighlightBuilder();
        HighlightBuilder.Field highlightAge = new HighlightBuilder.Field("name"); // name 字段高亮
        highlightAge.highlighterType("unified");  // 配置高亮类型
        highlightAge.preTags("<span style=\\\"color:red\\\">");
        highlightAge.postTags("</span>");
        highlightBuilder.field(highlightAge);  // 添加到 builder

        HighlightBuilder.Field highlightBirthday = new HighlightBuilder.Field("birthday");
        highlightBuilder.field(highlightBirthday);

        sourceBuilder.from(pageNum);
        sourceBuilder.size(pageSize);

        //使用QueryBuilders好像就无法高亮
//        QueryBuilder matchQueryBuilder = QueryBuilders.matchQuery("name", name)
//                // 启动模糊查询
//                .fuzziness(Fuzziness.AUTO)
//                // 在匹配查询上设置前缀长度选项
//                .prefixLength(3)
//                // 设置最大扩展选项以控制查询的模糊过程
//                .maxExpansions(10);
//        sourceBuilder.query(matchQueryBuilder);
        sourceBuilder.highlighter(highlightBuilder);
        searchRequest.source(sourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();
        //总命中数
        TotalHits totalHits = hits.getTotalHits();
        long total = totalHits.value;
        Map<String, Object> map = new HashMap<>();
        SearchHit[] searchHits = hits.getHits();
        map.put("count", total);
        List<Map<String, Object>> list = new ArrayList<>();
        for (SearchHit hit : searchHits) {
            //highlightFields.size=0??
            Map<String, HighlightField> highlightFields = hit.getHighlightFields();

            //name高亮
            HighlightField titleField = highlightFields.get("name");
            Map<String, Object> source = hit.getSourceAsMap();
            if (titleField != null) {
                Text[] fragments = titleField.fragments();
                String name2 = "";
                for (Text text : fragments) {
                    name2 += text;
                }
                source.put("name", name2);
            }
            list.add(source);
        }
        return list;
    }


    /**
     * bulkProcessor批量 异步
     *
     * @param portraitImageList
     * @return
     */
    //    @Override
    public BulkResponse portraitAdd3(List<PortraitImage> portraitImageList) {

        for (PortraitImage portraitImage : portraitImageList) {
            IndexRequest source = new IndexRequest("cluster").id(portraitImage.getPortraitId())
                    .source(JSON.toJSONString(portraitImage), XContentType.JSON);
            this.bulkProcessor.add(source);
        }

        return null;
    }
}